package com.thread.threadPool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author wangyingjie
 * @date
 * @remark 自定义线程池
 */
public class MyThreadPool {

    /**
     * 任务队列(需要执行的任务)
     * 因为线程池的实现其实是不断从任务列表中拉取任务去消耗，且需要在没任务时等待，而不是让线程死亡，所以用queue实现线程等待
     */
    private BlockingQueue<Runnable> tasks;

    /**
     * 存放工作线程的池子
     */
    private List<Work> threadPool;

    /**
     * 线程池是否中断工作：true中断， false不中断
     */
    private static volatile boolean shutDown = false;


    /**
     * 真正用于工作的线程
     */
    private class Work extends Thread {

        /**
         * 属于哪个线程池的标记
         */
        private MyThreadPool myThreadPool;

        public Work(MyThreadPool myThreadPool) {
            this.myThreadPool = myThreadPool;
        }

        /**
         * 工作线程的任务是从任务队列不停的拉去任务执行
         */
        @Override
        public void run() {
            while (!shutDown || tasks.size()>0){
                //线程池非中断情况或任务队列还有未消费任务，工作线程就要不停的去拉取任务
                try {
                    Runnable task;

                    if (!shutDown){
                        //由于在队列中没有任务时，线程需要等待阻塞，所以使用take
                        task = myThreadPool.tasks.take();
                        System.out.println("工作线程"+Thread.currentThread().getName()+"执行了");
                    }else {
                        //如果中断情况，工作线程拉取任务后，执行完自动结束，不阻塞
                        task = myThreadPool.tasks.poll();
                    }
                    if (task != null){
                        task.run();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    System.out.println("拉取任务队列报错");
                }
            }
        }
    }

    /**
     * 线程池初始化
     * @param taskSize 队列大小
     * @param poolSize 工作线程池大小
     */
    public MyThreadPool(int taskSize,int poolSize) {
        this.tasks = new ArrayBlockingQueue<>(taskSize);
        //初始化时需要同时创建工作线程池大小并放入工作线程
        this.threadPool = new ArrayList<>();
        for (int i = 0; i < poolSize; i++) {
            Work work = new Work(this);
            this.threadPool.add(work);
            work.start();
        }
        System.out.println("线程池开始初始化=====》"+"任务队列大小是："+taskSize+","+poolSize+"条工作线程被创建");
    }

    /**
     * 线程池执行任务（阻塞）
     * 从线程池取一条线程池去执行任务
     * @param task
     */
    public void submit(Runnable task){
        //将任务放入任务队列，让工作线程去消耗
        try {
            if (!shutDown){
                System.out.println("任务被提交到队列");
                //由于线程池中队列满时，需要让任务暂时等待阻塞，所以用put
                this.tasks.put(task);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    /**
     * 线程池执行任务（非阻塞）
     * 从线程池取一条线程池去执行任务
     * @param task
     */
    public void submitOffWait(Runnable task){
            System.out.println("任务被提交到队列");
            this.tasks.offer(task);
    }

    /**
     * 关闭线程池
     * 1.阻止添加新任务
     * 2.已有任务消耗光
     * 3.已有工作线程必须全部结束
     */
    public void shutDown() {
        shutDown =true;
        for (int i = 0; i < threadPool.size(); i++) {
            Work work = threadPool.get(i);
            Thread.State state = work.getState();
            if (state.equals(Thread.State.WAITING) ||
            state.equals(Thread.State.BLOCKED)){
                work.interrupt();
            }
        }
    }

    public static void main(String[] args) {
        MyThreadPool myThreadPool = new MyThreadPool(3, 6);
        for (int i = 0; i < 3; i++) {
            //提交3个任务
            myThreadPool.submit(()->{
                try {
                    Thread.sleep(2000);
                    System.out.println("任务内容是：xxx");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        //关闭线程池
        myThreadPool.shutDown();
    }

}
